package com.usoft.example.search;

import com.google.common.util.concurrent.SettableFuture;
import com.usoft.grpc.example.search.SearchRequest;
import com.usoft.grpc.example.search.SearchResponse;
import com.usoft.grpc.example.search.SearchServiceGrpc;
import io.grpc.ManagedChannel;
import io.grpc.ManagedChannelBuilder;
import io.grpc.stub.StreamObserver;

import java.util.Iterator;
import java.util.concurrent.TimeUnit;
import java.util.logging.Level;
import java.util.logging.Logger;

/**
 * Created by xinxingegeya on 15/9/25.
 */
public class SearchClient {

    private static final Logger logger = Logger
        .getLogger(SearchClient.class.getName());

    private final ManagedChannel channel;
    private final SearchServiceGrpc.SearchServiceBlockingStub blockingStub;
    private final SearchServiceGrpc.SearchServiceStub asyncStub;


    /**
     * Construct client connecting to HelloWorld server at {@code host:port}.
     */
    public SearchClient(String host, int port) {
        channel = ManagedChannelBuilder.forAddress(host, port)
            .usePlaintext(true).build();
        blockingStub = SearchServiceGrpc.newBlockingStub(channel);
        asyncStub = SearchServiceGrpc.newStub(channel);
    }

    public void shutdown() throws InterruptedException {
        channel.shutdown().awaitTermination(5, TimeUnit.SECONDS);
    }

    /**
     * simple rpc
     * 
     * @param pageNo
     * @param pageSize
     */
    public void searchWithSimpleRpc(int pageNo, int pageSize) {

        try {
            logger.info(
                "search param pageNo=" + pageNo + ",pageSize=" + pageSize);
            SearchRequest request = SearchRequest.newBuilder()
                .setPageNumber(pageNo).setResultPerPage(pageSize).build();
            SearchResponse response = blockingStub.searchWithSimpleRpc(request);
            logger.info("search result: " + response.toString());
        } catch (RuntimeException e) {
            logger.log(Level.WARNING, "RPC failed", e);
            return;
        }
    }

    /**
     * server side stream rpc
     * 
     * @param pageNo
     * @param pageSize
     */
    public void searchWithSeverSideStreamRpc(int pageNo, int pageSize) {
        try {
            logger.info(
                "search param pageNo=" + pageNo + ",pageSize=" + pageSize);
            SearchRequest request = SearchRequest.newBuilder()
                .setPageNumber(pageNo).setResultPerPage(pageSize).build();
            Iterator<SearchResponse> responseIterator = blockingStub
                .searchWithServerSideStreamRpc(request);
            while (responseIterator.hasNext()) {
                SearchResponse r = responseIterator.next();
                if (r.getResult(0).getSnippets(0).equals("the last")) {
                    logger.info("the end: \n" + r.toString());
                    break;
                }
                logger.info("search result:\n " + r.toString());
            }
        } catch (RuntimeException e) {
            logger.log(Level.WARNING, "RPC failed", e);
            return;
        }
    }

    /**
     * client side stream rpc
     * 
     * @param pageNo
     * @param pageSize
     * @throws Exception
     */
    public void searchWithClientSideStreamRpc(int pageNo, int pageSize)
            throws Exception {
        final SettableFuture<Void> finishFuture = SettableFuture.create();
        StreamObserver<SearchResponse> responseObserver = new StreamObserver<SearchResponse>() {
            @Override
            public void onNext(SearchResponse searchResponse) {
                logger.info(
                    "response with result=\n" + searchResponse.toString());
            }

            @Override
            public void onError(Throwable throwable) {
                finishFuture.setException(throwable);
            }

            @Override
            public void onCompleted() {
                finishFuture.set(null);
            }
        };
        StreamObserver<SearchRequest> requestObserver = asyncStub
            .searchWithClientSideStreamRpc(responseObserver);
        try {

            // 发送三次search request
            for (int i = 1; i <= 3; i++) {
                SearchRequest request = SearchRequest.newBuilder()
                    .setPageNumber(pageNo).setResultPerPage(pageSize + i)
                    .build();
                requestObserver.onNext(request);
                if (finishFuture.isDone()) {
                    logger.log(Level.WARNING, "finish future is done");
                    break;
                }
            }
            requestObserver.onCompleted();
            finishFuture.get();
            logger.log(Level.INFO, "finished");
        } catch (Exception e) {
            requestObserver.onError(e);
            logger.log(Level.WARNING, "Client Side Stream Rpc Failed", e);
            throw e;
        }
    }

    /**
     * bidirectional stream rpc
     * 
     * @param pageNo
     * @param pageSize
     */
    public void searchWithBidirectionalStreamRpc(int pageNo, int pageSize)
            throws Exception {
        final SettableFuture<Void> finishFuture = SettableFuture.create();

        StreamObserver<SearchRequest> requestObserver = asyncStub
            .searchWithBidirectionalStreamRpc(
                new StreamObserver<SearchResponse>() {
                    @Override
                    public void onNext(SearchResponse searchResponse) {
                        logger.info("response with result = \n"
                            + searchResponse.toString());
                    }

                    @Override
                    public void onError(Throwable throwable) {
                        finishFuture.setException(throwable);
                    }

                    @Override
                    public void onCompleted() {
                        finishFuture.set(null);
                    }
                });
        try {
            // 发送三次search request
            for (int i = 1; i <= 3; i++) {
                SearchRequest request = SearchRequest.newBuilder()
                    .setPageNumber(pageNo).setResultPerPage(pageSize + i)
                    .build();
                requestObserver.onNext(request);
            }
            requestObserver.onCompleted();
            finishFuture.get();
            logger.log(Level.INFO, "finished");
        } catch (Exception e) {
            requestObserver.onError(e);
            logger.log(Level.WARNING, "Bidirectional Stream Rpc Failed", e);
            throw e;
        }
    }

    /**
     * client
     */
    public static void main(String[] args) throws Exception {
        SearchClient client = new SearchClient("localhost", 50051);
        try {
//            client.searchWithSimpleRpc(1, 13);
//            client.searchWithSeverSideStreamRpc(1, 2);
//            client.searchWithClientSideStreamRpc(1, 3);
            client.searchWithBidirectionalStreamRpc(1, 3);
        } finally {
            client.shutdown();
        }
    }
}
